package org.midy.infrastructure;

import org.midy.infrastructure.customer.OnlineList;
import org.midy.infrastructure.tools.CloseHandler;
import org.midy.infrastructure.tools.Logger;
import org.midy.infrastructure.tools.SelectorTools;

import java.io.Closeable;
import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Objects;
import java.util.OptionalInt;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Worker 模型:
 * 1. 一个线程就是一个Worker
 * 2. 一个Worker有且仅维护一个Selector
 * 3. 一个Selector负责网络事件的监听、IO读写、消息路由
 */
public class IOEventDrivenWorker implements Closeable, Runnable {
    /* 当前"工作者"的名字 */
    private final String workerName;
    private final Selector selector;
    private final Thread workerThread;
    /* 当前Worker待注册的监听连接事件 */
    private final BlockingQueue<UnregisteredTask> ioEventTaskListenUnProcessQueue = new LinkedBlockingQueue<>();
    /* 状态管理器，用来协作和标识Worker的状态 */
    private final AtomicBoolean status = new AtomicBoolean(true);
    /* 事件处理器 */
    private final ReadEventHandler readEventHandler;
    private final AcceptEventHandler acceptEventHandler;
    private final ConnectEventHandler connectEventHandler;
    private final WriteEventHandler writeEventHandler;
    /* 接收到的信息要派发 */
    private final InformationDistribution informationDistribution;
    /**
     * 其他控制参数部分
     */

    /* 空闲自检周期, 单位：毫秒 */
    private long selectTimeout = 60000;

    /**
     * @param workerName                      线程名称关联到容器
     * @param informationDistribution   读取到消息需要总线分发，该接口提供总线分发能力
     * @param eventDrivenTaskAllocation 关联另外一个事件驱动任务分配器
     */
    public IOEventDrivenWorker(String workerName, InformationDistribution informationDistribution, EventDrivenTaskAllocation eventDrivenTaskAllocation) {
        this.workerName = workerName;
        /* 新连接需要监听的新任务 */
        this.informationDistribution = informationDistribution;
        /* 内置事件处理器：连接事件、监听事件、读事件、写事件 */
        this.acceptEventHandler = new AcceptEventHandler(eventDrivenTaskAllocation);
        this.connectEventHandler = new ConnectEventHandler(eventDrivenTaskAllocation);
        this.readEventHandler = new ReadEventHandler();
        this.writeEventHandler = new WriteEventHandler();
        /* 一个线程维护一个 selector, 避免并发 */
        this.selector = initSelector();
        /* 初始化一个线程且唯一绑定一个Selector */
        this.workerThread = new Thread(this, this.workerName);
        this.workerThread.setDaemon(true);
        this.workerThread.start();
    }

    /**
     * 初始化一个Selector
     *
     * @return Selector
     */
    public Selector initSelector() {
        try {
            return Selector.open();
        } catch (Throwable throwable) {
            Logger.error("初始化Selector失败", throwable);
            throw new RuntimeException(throwable);
        }
    }

    /**
     * Worker - Thread 循环事件处理逻辑
     */
    @Override
    public void run() {
        Logger.info("[%s] Worker 开始业务处理 ", workerName);
        try {
            while (status.get()) {
                int select = 0;
                try {
                    try {
                        select = selector.select(selectTimeout);
                        if (select > 0) {
                            Set<SelectionKey> selectionKeys = selector.selectedKeys();
                            newUnProcessEvent(selectionKeys, selector);
                        }
                    } catch (Throwable t) {
                        Logger.error("循环事件处理失败", t);
                    }
                    try {
                        unRegisterTaskEvent();
                    } catch (Throwable throwable) {
                        Logger.error("待注册事件处理失败", throwable);
                    }

                } finally {
                    Logger.info("[COMPLETE] 总事件数：%d, 本轮处理的事件数：%d, ", selector.keys().size(), select);
                }
            }

        } finally {
            /* 线程任务退出, 需要清理资源 */
            closeSelector();
            closeWorkerQueue();
            Logger.info("[%s] Worker 业务处理已退出", workerName);
        }
    }

    /**
     * 待处理的IO事件
     *
     * @param selectionKeys 事件状态key
     */
    private void newUnProcessEvent(Set<SelectionKey> selectionKeys, Selector selector) {
        if (selectionKeys == null || selectionKeys.isEmpty()) {
            return;
        }

        Iterator<SelectionKey> iterator = selectionKeys.iterator();
        while (iterator.hasNext()) {
            SelectionKey selectionKey = null;
            try {
                selectionKey = iterator.next();
                /* 无效Key的事件直接跳过 */
                if (!selectionKey.isValid()) {
                    continue;
                }
                /* 读事件处理 */
                if (selectionKey.isReadable()) {
                    byte[] readMessage = (byte[]) readEventHandler.handle(selectionKey);
                    /* readMessage is null 的场景下不需要消息进行总线路由 */
                    if (readMessage != null) {
                        informationDistribution.routeData(readMessage, (SocketChannel) selectionKey.channel(), selectionKey, this);
                    }
                    /* 写数据事件： 写数据模型为：先记录到对应网络连接的写消息队列中，并给当前网络连接注册一个写事件 */
                } else if (selectionKey.isWritable()) {
                    writeEventHandler.handle(selectionKey);
                    /* 服务端新的连接事件：接受新建立连接请求 */
                } else if (selectionKey.isAcceptable()) {
                    acceptEventHandler.handle(selectionKey);
                    /* 连接完成事件：接受新建立连接请求 */
                } else if (selectionKey.isConnectable()) {
                    connectEventHandler.handle(selectionKey);
                } else {
                    Logger.warn("无效的事件!");
                    selectionKey.cancel();
                }
                /*
                 * 出现位置异常得兜底机制：
                 *  1. 关闭连接
                 *  2. 释放所有资源：
                 *      2.1 所有的attach数据     释放当前连接的所有缓存资源
                 *      2.2 selectionKey       关闭当前连接对应的网路事件
                 *      2.3 SelectableChannel  关闭网络连接
                 */
            } catch (Throwable t1) {
                Logger.warn("网络事件处理警告: %s", t1.getMessage());
                closeAfterChannelException(selectionKey);
                logout(selectionKey);
            } finally {
                // 移除已经处理完毕的时间
                iterator.remove();
            }
        }
    }

    private void logout(SelectionKey selectionKey) {
        OnlineList.logout(selectionKey.channel());
    }

    private void login(SelectionKey selectionKey) {
        SelectableChannel channel = selectionKey.channel();
        if(channel instanceof SocketChannel) {
            try {
                String ip_port = ((SocketChannel) channel).getRemoteAddress().toString();
                OnlineList.registerOnline(ip_port, selectionKey.channel(), selectionKey, this);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public void closeAfterChannelException(SelectionKey selectionKey) {
        if (Objects.nonNull(selectionKey)) {
            SelectableChannel orderChannel = selectionKey.channel();
            if (orderChannel.isOpen()) {
                try {
                    orderChannel.close();
                } catch (Throwable t2) {
                    Logger.error("网络事件处理失败, 关闭连接异常！", t2);
                }
            }

            Object attach = null;

            if (selectionKey.isValid()) {
                attach = selectionKey.attach(null);
                selectionKey.cancel(); // 取消当前连接的所有的网络事件监听
            }

            if(attach instanceof ChannelEventHelper) {
                ChannelEventHelper channelEventHelper = (ChannelEventHelper) attach;
                channelEventHelper.close();
            }
        }
    }

    /**
     * 当前循环事件组处理待监听事件的连接
     */
    private void unRegisterTaskEvent() {
        UnregisteredTask task = null;
        while ((task = ioEventTaskListenUnProcessQueue.poll()) != null) {
            try {
                if (!task.socketChannel.isOpen()) {
                    // 当前连接是关闭的不做任何处理
                    Logger.warn("网络连接已经关闭!");
                    continue;
                }
                /* 在原来的SelectionKey上追加监听事件: 需要识别出，当前SocketChannel在当前Selector上有监听事件吗
                 * 情况1： 给过来的SelectionKey是一个属于当前Worker的有效事件监听key
                 * 情况2： 从当前Selector中通过SocketChannel查询SelectionKey
                 * 问题点： 通过SocketChannel 找SelectionKey 是一个随连接数性能下降的点
                 */
                SelectionKey existSelectionKey = null;
                if (Objects.nonNull(existSelectionKey = task.selectionKey) && existSelectionKey.isValid() && existSelectionKey.selector() == this.selector ||
                        Objects.nonNull(existSelectionKey = task.socketChannel.keyFor(this.selector)) && existSelectionKey.isValid()) {
                    ChannelEventHelper orderAttachment = (ChannelEventHelper) existSelectionKey.attachment();

                    if (Objects.isNull(orderAttachment)) {
                        orderAttachment = ChannelEventHelper.attachment(existSelectionKey);
                    }

                    if (Objects.nonNull(task.writeData)) {
                        orderAttachment.putSendByteBuffer(ByteBuffer.wrap(task.writeData));
                    }

                    if ((existSelectionKey.interestOps() & task.unSelectionKey) != task.unSelectionKey) {
                        existSelectionKey.interestOps(existSelectionKey.interestOps() | task.unSelectionKey);
                    }
                } else {
                    ChannelEventHelper channelEventHelper = ChannelEventHelper.buildAttachment(task.socketChannel);
                    if (Objects.nonNull(task.writeData)) {
                        channelEventHelper.putSendByteBuffer(ByteBuffer.wrap(task.writeData));
                    }
                    SelectionKey register = task.socketChannel.register(this.selector, task.unSelectionKey, channelEventHelper);
                    channelEventHelper.setSelectionKey(register);
                    login(register);// 注册登录
                }
            } catch (Throwable throwable) { // 这种情况目前
                Logger.error("处理待监听连接时产生异常", throwable);
            } finally {
                String address = "";
                try {
                    if(task.socketChannel instanceof SocketChannel) {
                        SocketChannel sc = (SocketChannel) task.socketChannel;
                        SocketAddress remoteAddress = sc.getRemoteAddress();
                        if (remoteAddress != null) {
                            address = remoteAddress.toString();
                        }
                    }
                } catch(Throwable throwable) {
                    Logger.error("获取IP地址信息失败", throwable);
                }

                Logger.info("address：%s, 事件类型：%s", address, SelectorTools.parse(task.unSelectionKey));
            }
        }
    }

    /**
     * 给
     *
     * @param task
     * @param unSelectionKey
     * @return
     */

    public boolean register(SelectableChannel task, int unSelectionKey) {
        UnregisteredTask unProcessTask = new UnregisteredTask(task, unSelectionKey);
        boolean offer = ioEventTaskListenUnProcessQueue.offer(unProcessTask);
        if (offer) {
            selector.wakeup();
        }
        return offer;
    }

    public boolean register(SelectableChannel task, SelectionKey selectionKey, int unSelectionKey, byte[] writeDate) {
        UnregisteredTask unProcessTask = new UnregisteredTask(task, selectionKey, unSelectionKey, writeDate);
        boolean offer = ioEventTaskListenUnProcessQueue.offer(unProcessTask);
        if (offer) {
            selector.wakeup();
        }
        return offer;
    }

    @Override
    public void close() throws IOException {
        while (status.get()) {
            if (status.compareAndSet(true, false)) {
                try {
                    closeSelector();
                    closeWorkerQueue();
                    workerThread.interrupt();
                } finally {
                    break;
                }
            }
        }

    }

    private void closeSelector() {
        CloseHandler.closeSelector(selector);
    }

    /**
     * 需要关闭待监听事件的网络简介
     */
    private void closeWorkerQueue() {
        if (Objects.isNull(ioEventTaskListenUnProcessQueue) || ioEventTaskListenUnProcessQueue.isEmpty()) {
            return;
        }

        UnregisteredTask task = null;
        while ((task = ioEventTaskListenUnProcessQueue.poll()) != null) {
            try {
                if (task.socketChannel.isOpen()) {
                    task.socketChannel.close();
                }
            } catch (Throwable throwable) {
                Logger.error("关闭待处理队列时，产生失败", throwable);
            }
        }
    }

    public String getWorkerName() {
        return workerName;
    }

    public long getSelectTimeout() {
        return selectTimeout;
    }

    public void setSelectTimeout(long selectTimeout) {
        this.selectTimeout = selectTimeout;
    }

    static class UnregisteredTask {

        public SelectionKey selectionKey;

        public SelectableChannel socketChannel;

        public int unSelectionKey;

        public byte[] writeData;

        public UnregisteredTask() {
        }

        public UnregisteredTask(SelectableChannel socketChannel, int unSelectionKey) {
            this.socketChannel = socketChannel;
            this.unSelectionKey = unSelectionKey;
        }

        public UnregisteredTask(SelectableChannel socketChannel, SelectionKey selectionKey, int unSelectionKey, byte[] writeData) {
            this.socketChannel = socketChannel;
            this.unSelectionKey = unSelectionKey;
            this.selectionKey = selectionKey;
            this.writeData = writeData;
        }
    }


}
